package com.yutian.socketnio.server;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.nio.ByteBuffer;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Set;

/**
 * https://www.jianshu.com/p/a9d030fec081
 * https://blog.csdn.net/jeffleo/article/details/54695959
 * nio：io多路复用，即一个线程可以负责多个请求。有个selector（池）负责接收连接事件，分配线程（池）处理！
 */
public class HttpNioServer {
    private static Logger log = LoggerFactory.getLogger(HttpNioServer.class);

    public void listen(int port) {
        ServerSocketChannel serverChannel = null;
        ServerSocket ss = null;
        Selector selector = null;
        try{
            //打开Selector来处理 Channel
            selector = Selector.open();

            serverChannel = ServerSocketChannel.open();
            serverChannel.configureBlocking(false);
            ss = serverChannel.socket();
            InetSocketAddress address = new InetSocketAddress(port);
            //将服务器绑定到选定的端口
            ss.bind(address);

            //将ServerSocket注册到Selector以接受连接
            //一旦我们向Selector注册了一个或多个channel（即ServerSocket）后，就可以调用select来获取channel。select方法会返回所有处于就绪状态的channel
            serverChannel.register(selector, SelectionKey.OP_ACCEPT);

            while (!Thread.currentThread().isInterrupted()) {
                //选择一组键，其相应的通道已为 I/O 操作准备就绪。
                try {
                    //等待需要处理的新事件；阻塞将一直持续到下一个传入事件
                    //select()方法的返回值是一个int整形，代表有多少channel处于就绪了。也就是自上一次select后有多少channel进入就绪。
                    int readyChannels = selector.select();
                    if(readyChannels == 0) continue;
                } catch (IOException ex) {
                    log.error(ex.getMessage(), ex);
                    //handle exception
                    break;
                }
                //获取所有接收事件的SelectionKey实例
                Set<SelectionKey> readyKeys = selector.selectedKeys();
                Iterator<SelectionKey> iterator = readyKeys.iterator();
                while (iterator.hasNext()) {
                    SelectionKey key = iterator.next();
                    if (!key.isValid() /*&& key.isConnectable()*/) {
                        continue;
                    }
                    handleSelectionKey(selector, key);
                    //该事件已经处理，可以丢弃
                    iterator.remove(); // Selector本身并不会移除SelectionKey对象。当下次channel处于就绪时，Selector仍然会把这些key再次加入进来。
                }
            }
            // 当操作Selector完毕后，需要调用close方法。close的调用会关闭Selector并使相关的SelectionKey都无效。channel本身不管被关闭。
            selector.close();
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
    }

    /**
     * 扩展：分配线程（池）异步处理
     * @param selector
     * @param key
     */
    private void handleSelectionKey(Selector selector, SelectionKey key) {
        if (key.isAcceptable()) {
            accept(selector, key);
        } else if (key.isReadable()) {
            read(selector, key);
        } else if (key.isWritable()) {
            write(selector, key);
        }
    }

    private void write(Selector selector, SelectionKey key) {
        SocketChannel channel = (SocketChannel) key.channel();

        try {
            // 模拟业务逻辑耗时
            Thread.sleep(1000L);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        String str = htmlContent("Hi，我来自服务端的问候！");
        log.info("write:"+str);
        ByteBuffer sendBuffer = ByteBuffer.allocate(1024);//调整缓存的大小可以看到打印输出的变化
        sendBuffer.clear();
        sendBuffer.put(str.getBytes());
        sendBuffer.flip();
        try {
            channel.write(sendBuffer);
            channel.register(selector, SelectionKey.OP_READ);
        } catch (IOException e) {
            log.error(e.getMessage(), e);
        }
    }

    /**
     * 无论什么请求，都返回html
     * @param content
     * @return
     */
    private static String htmlContent(String content){
        return "HTTP/1.1 200 OK\r\n" +
                "Content-Type: text/html; charset=utf-8\r\n" +
                "\r\n" +
                "<html><body><h1>"+content+"</h1></body></html>\r\n";
    }

    private void read(Selector selector, SelectionKey key) {
        log.info("receive message...");
        // SelectionKey.channel返回的channel实例需要强转为我们实际使用的具体的channel类型
        SocketChannel socketChannel = (SocketChannel) key.channel();
        ByteBuffer readBuffer = ByteBuffer.allocate(1024);//调整缓存的大小可以看到打印输出的变化
        // Clear out our read buffer so it's ready for new data 
        // readBuffer.clear();
//        readBuffer.flip();
        // Attempt to read off the channel 
        int numRead;
        try {
            numRead = socketChannel.read(readBuffer);
        } catch (IOException e) {
            // 客户端断开连接，FD_READ仍然起作用，但读取异常，所以从Selector中取消注册  TODO 未确认
            // The remote forcibly closed the connection, cancel 
            // the selection key and close the channel. 
            closeChannel(key, socketChannel);
            log.error("socketChannel.read.error", e);
            return;
        }
        if(-1 == numRead){
            // 客户端主动关闭连接，服务器端并不会出现远程主机强迫关闭一个现有的连接这样的异常，而是会一直出现可读事件，一直返回-1。已验证！
            closeChannel(key, socketChannel);
            log.error("socketChannel.read.-1");
            return;
        }
        try {
            // 模拟业务逻辑耗时
            Thread.sleep(1000L);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        String str = new String(readBuffer.array(), 0, numRead);
        log.info("read:"+str);
        try {
            socketChannel.register(selector, SelectionKey.OP_WRITE);
        } catch (ClosedChannelException e) {
            log.error(e.getMessage(), e);
        }
    }

    private void closeChannel(SelectionKey key, SocketChannel socketChannel) {
        key.cancel();
        try {
            socketChannel.close();
        } catch (IOException ex) {
            log.error(ex.getMessage(), ex);
        }
    }

    // 当 OP_ACCEPT 事件到来时, 我们就有从 ServerSocketChannel 中获取一个 SocketChannel,
    // 代表客户端的连接
    // 注意, 在 OP_ACCEPT 事件中, 从 key.channel() 返回的 Channel 是 ServerSocketChannel.
    // 而在 OP_WRITE 和 OP_READ 中, 从 key.channel() 返回的是 SocketChannel.
    private void accept(Selector selector, SelectionKey key) {
        ServerSocketChannel ssc = (ServerSocketChannel) key.channel();
        try {
            SocketChannel clientChannel = ssc.accept();
            clientChannel.configureBlocking(false);
            clientChannel.register(selector, SelectionKey.OP_READ);
            log.info("a new client connected "+clientChannel.getRemoteAddress());
        } catch (IOException e) {
            log.error(e.getMessage(), e);
        }
    }

}